package syM

import (
	"db2s/arg"
	cp "db2s/componentInit"
	"db2s/global"
	"db2s/go-log/log"
	"db2s/pubM"
	quit "db2s/quitDispos"
	"db2s/ref"
	mq "db2s/topic-mq"
	"errors"
	"fmt"
	"github.com/panjf2000/ants/v2"
	"github.com/vbauerster/mpb/v8"
	"os"
	"reflect"
	"sync"
	"sync/atomic"
	"time"
)

var (
	monitorListener mq.MonitorListener
)

type sendTopicF func(check CheckMode, topic Topic, tableName string, msg map[string]any) bool
type sendEndAction func()

func GetCheckSchemaMetaData() *CheckSchemaMetaData {
	return &CheckSchemaMetaData{TableObjectFlow: make(chan any, global.ChannelDefault)}
}
func TableCountActive(TableObject []*arg.InputTableAttributes) (sum int) {
	return len(TableObject)
}
func newSchemaInitContext(seq int, v arg.InputTableAttributes, logSeq int64) *SchemaInitContext {
	return &SchemaInitContext{
		Schema: v.Schema, Table: v.Table, MSchema: v.MSchema,
		MTable: v.MTable, Seq: seq, ObjectLogSeq: logSeq,
		Role: v.TableAttributes.Role, TableAttributes: v.TableAttributes,
	}
}
func (t CheckSchemaMetaData) objectMetaInitAbnormalQuit(seq int, v arg.InputTableAttributes, res *SchemaInitContext) bool {
	var err error
	if res.TPod, err = t.podInit(seq, v); err != nil {
		return true
	}
	if atomic.LoadInt64(v.TableStatus) == -1 {
		res.TPod.TableMiss = true
		return true
	}
	return false
}
func (t CheckSchemaMetaData) objectMetaInit(seq int, v arg.InputTableAttributes, logSeq int64) (res *SchemaInitContext, err error) {
	res = newSchemaInitContext(seq, v, logSeq)
	if t.objectMetaInitAbnormalQuit(seq, v, res) {
		res.TableMiss = true
		return
	}
	return
}
func schemaMetaExecFuncParameterSplit(f []any) (baseParameter map[string]any, chanParameter any, useParameter []any) {
	baseParameter = f[0].(map[string]any)
	chanParameter = f[1]
	useParameter = f[2:]
	return
}
func (t CheckSchemaMetaData) ChanSchemaContextObjectInit() *SchemaInitContext {
	var tableFree, tableAccount, breakPointSum int64
	return &SchemaInitContext{ChanSchemaContext: make(chan any, t.ObjectSum), TableSum: t.ObjectSum, IgnoreObjectSum: t.IgnoreObjectSum, TableAccount: &tableAccount, TableFree: &tableFree, BreakPointSum: &breakPointSum}
}
func (t CheckSchemaMetaData) schemaMetaExecFunc(_ string, f any) (r any) {
	if f == nil || len(f.([]any)) == 0 {
		return
	}
	baseParameter, chanParameter, userParameter := schemaMetaExecFuncParameterSplit(f.([]any))
	if _, ok := baseParameter["seq"]; !ok {
		return
	}
	if _, ok := baseParameter["logSeq"]; !ok {
		return
	}
	seq := baseParameter["seq"].(int)
	logSeq := baseParameter["logSeq"].(int64)
	p, err := t.objectMetaInit(seq, chanParameter.(arg.InputTableAttributes), logSeq)
	if err != nil {
		return
	}
	ChanSchemaContext := userParameter[0].(*SchemaInitContext)
	atomic.AddInt64(ChanSchemaContext.TableAccount, 1)
	atomic.AddInt64(ChanSchemaContext.TableFree, 1)
	ChanSchemaContext.ChanSchemaContext <- p
	return
}
func (t CheckSchemaMetaData) Check() {
	var event = "[MetaCheck]"
	continueTableList = breakP.GetRes("result").(map[string]int)
	ChanSchemaContext := t.ChanSchemaContextObjectInit()
	fmt.Println(fmt.Sprintf("-- [%v] %v report: begin init per table object group metadata -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
	log.MainLog().Info(fmt.Sprintf("%v Start initializing table object data", event))
	go func() {
		defer func() {
			close(ChanSchemaContext.ChanSchemaContext)
		}()
		if err := pubM.ObjectMetaDataInit(pubM.ObjectMetaDataPoint{
			NewBreakPoint: pubM.NewBreakPoint{
				BreakPointSum:     ChanSchemaContext.BreakPointSum,
				ContinueTableList: continueTableList,
				ObjectFlow:        t.TableObjectFlow,
				TableAccount:      ChanSchemaContext.TableAccount,
				TableFree:         ChanSchemaContext.TableFree,
			},
			ExecFunc: global.EFunc{FuncName: t.schemaMetaExecFunc, Params: []any{ChanSchemaContext}},
		}).NormalChanSendMsg("syRCheck"); err != nil {
			return
		}
	}()
	global.GetIoc().RegisterContext("schemaInitContext", ChanSchemaContext)
}

func FuncExportVar() {
	//函数调用映射
	FuncRef = global.GetIoc().GetBean("checkObjectFuncContext").(map[string]reflect.Value)
	ScheduleTaskInitFuncRef = global.GetIoc().GetBean("scheduleTaskInitFuncContext").(map[string]reflect.Value)
	closeObject = global.GetIoc().GetBean("closeObject").(quit.Close)
	MqListenerSInitFunc = global.GetIoc().GetBean("mqListenerSInitFuncContext").(map[string]reflect.Value)
}

// ExportVar 导出变量 方便使用
func ExportVar() {
	//producer 生产者
	producer = global.GetIoc().GetBean("producer").(mq.Producer)
	//consumer 消费者
	consumer = global.GetIoc().GetBean("consumer").(mq.Consumer)
	//monitorListener mq 监控
	monitorListener = global.GetIoc().GetBean("monitorListener").(mq.MonitorListener)
	PlanContext = global.GetIoc().GetBean("schedulePlanContext").(*SchedulePlanContext) //注册ioc 容器
}
func (context *SchedulePlanContext) GetIndexMode() CheckMode {
	switch GlobalPConfigs.rules.CheckMode {
	case "struct":
		return "struct"
	case "object":
		return "object"
	case "foreign":
		return "foreign"
	case "func":
		return "func"
	case "proc":
		return "proc"
	case "trigger":
		return "trigger"
	}
	return ""
}
func getCheckMod() CheckMode {
	switch GlobalPConfigs.rules.CheckMode {
	case "struct":
		return PlanContext.GetIndexMode()
	case "object":
		return PlanContext.GetIndexMode()
	case "foreign":
		return PlanContext.GetIndexMode()
	case "func":
		return PlanContext.GetIndexMode()
	case "proc":
		return PlanContext.GetIndexMode()
	case "trigger":
		return PlanContext.GetIndexMode()
	}
	return ""
}

func (context *SchedulePlanContext) objectEndMark(index CheckMode) {
	context.AutoCloseMqComponentTable[cp.CheckMode(index)].ObjectEndMark()
}
func structToTopic(plan *SchedulePlan2) {
	sendMsg(getCheckMod(), StructIsBaseQuery, getTableName(plan.Object.Schema, plan.Object.Table), map[string]any{"segmentSql": map[string]string{}, "plan": plan, "subTask": plan.subTaskInfo, "topic": string(StructIsBaseQuery)})
}
func objectToTopic(plan *SchedulePlan2) {
	sendMsg(getCheckMod(), ObjectBaseQuery, getTableName(plan.Object.Schema, plan.Object.Table), map[string]any{"segmentSql": map[string]string{}, "plan": plan, "subTask": plan.subTaskInfo, "topic": string(ObjectBaseQuery)})
}

func checkObjectFunc() map[string]reflect.Value {
	return map[string]reflect.Value{
		"struct": reflect.ValueOf(structToTopic),
		"object": reflect.ValueOf(objectToTopic),
	}
}
func sendScheduleMsgFunc() map[string]reflect.Value {
	return map[string]reflect.Value{
		"struct": reflect.ValueOf(initStructSchedulePlan),
		"object": reflect.ValueOf(initObjectSchedulePlan),
	}
}
func mqListenerSInitFunc() map[string]reflect.Value {
	return map[string]reflect.Value{
		"struct": reflect.ValueOf(structMQListenInitFunc),
		"object": reflect.ValueOf(objectMQListenInitFunc),
	}
}

func schedulePlanGetIocInit() {
	global.GetIoc().RegisterContext("checkObjectFuncContext", checkObjectFunc())
	global.GetIoc().RegisterContext("scheduleTaskInitFuncContext", sendScheduleMsgFunc())
	global.GetIoc().RegisterContext("mqListenerSInitFuncContext", mqListenerSInitFunc())
	FuncExportVar()
}
func schedulePlanThreadPoolInit() (pool *ants.Pool, err error) {
	initContext := global.GetIoc().GetBean("schemaInitContext").(*SchemaInitContext)
	//初始化上下文线程池
	if pool, err = ants.NewPool(initContext.TableSum); err != nil {
		return
	}
	progress := mpb.New(mpb.WithWidth(80))
	global.GetIoc().RegisterContext("progress", progress)
	return
}
func planContextBarInit(config *SchedulePlanContextConfig) *mq.TaskBar {
	initContext := global.GetIoc().GetBean("schemaInitContext").(*SchemaInitContext)
	return cp.InitBarMonitor(cp.Bar{
		TaskTotal:     initContext.TableAccount,
		Task:          initContext.TableSum,
		Free:          initContext.TableFree,
		BreakPointSum: initContext.BreakPointSum,
		CheckMod:      config.GtCheckConfig.SecondaryL.RulesV.CheckMode,
		Object:        config.GtCheckConfig.SecondaryL.SchemaV.CheckIndexObject,
		BeginTime:     config.BeginTime,
		TtyBarFin:     GlobalPConfigs.FilePoint.TtyBarFin,
	}, len(GetCheckToTopicMQListeners()[cp.CheckMode(config.GtCheckConfig.SecondaryL.RulesV.CheckMode)]))
}

func PlanContextInit(config *SchedulePlanContextConfig) (PlanContext *SchedulePlanContext, err error) {
	var pool *ants.Pool
	//初始化上下文
	if pool, err = schedulePlanThreadPoolInit(); err != nil {
		return
	}
	PlanContext = &SchedulePlanContext{
		Pool:                      pool,
		AutoCloseMqComponentTable: cp.InitAutoCloseMqComponentTable(),
		BeginTime:                 config.BeginTime,
	}
	PlanContext.TaskBarTable = planContextBarInit(config) //初始化执行计划表
	PlanContext.ChanSchedulePlanTable = InitToSchedulePlan2Table()
	global.GetIoc().RegisterContext("schedulePlanContext", PlanContext) //注册ioc 容器
	return
}

// NewSchedulePlanContext 创建执行计划上下文
func schedulePlanContext(config *SchedulePlanContextConfig) (PlanContext1 *SchedulePlanContext) {
	var (
		err error
	)
	fmt.Println(fmt.Sprintf("-- [%v] %v report: begin init schedule task Connext -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
	lock.Lock()
	defer lock.Unlock()
	schedulePlanGetIocInit()
	if err = initContextBean(config); err != nil {
		fmt.Println("创建执行计划上下文 NewSchedulePlanContext 失败", err)
		return nil
	}
	if PlanContext1, err = PlanContextInit(config); err != nil {
		return nil
	}
	//导出变量
	ExportVar()
	return
}

// initContextBean 初始化上下文需要用到的组件
func initContextBean(config *SchedulePlanContextConfig) error {
	//TODO: checkConfig 行哥 你来写
	createMq := config.MQConfig.CreateMq(GetCheckToTopicMQListeners())
	if createMq == nil {
		return errors.New("初始化 mq 失败")
	}
	return nil
}
func (context *SchedulePlanContext) Close() {
	for _, component := range context.AutoCloseMqComponentTable {
		context.wg.Add(1)
		func() {
			defer context.wg.Done()
			component.Close(false)
		}()
	}
	context.wg.Wait()
}

//	func breakPointQuit(event string, _ any) (r any) {
//		ref.MyWaitDel("breakPoint")
//		log.MainLog().Info(fmt.Sprintf("%v When the breakpoint resume function is enabled, table set %v will be skipped.", "getBreakPointTableList", continueTableList))
//		return
//	}
//
//	func breakPointList(event string, f any) (r any) {
//		if f == nil || len(f.([]any)) == 0 {
//			return
//		}
//		_, chanParameter, _ := schemaMetaExecFuncParameterSplit(f.([]any))
//		switch chanParameter.(type) {
//		case *string:
//			continueTableList[*chanParameter.(*string)]++
//		}
//		return
//	}
//
//	func breakPointObjectInit(m any) (send *mq.MonitorMsgMq) {
//		send = mq.NewMonitorMsgMq()
//		send.SendMsg.MsgModifyKey = m.(chan any)
//		send.NormalQuit = global.EFunc{FuncName: breakPointQuit}
//		send.ExecFunc = append(send.ExecFunc, global.EFunc{FuncName: breakPointList})
//		return
//	}
//
//	func getBreakPointTableList() (err error) {
//		var m any
//		if !getBreakPointStatus(*GlobalPConfigs) {
//			return
//		}
//		if m, err = newTextBreakPoint.Read(); err != nil {
//			log.ErrorLog().Error(fmt.Sprintf("Failed to read the breakpoint file %v. error is %v", GlobalPConfigs.FilePoint.BreakPointFin.Name(), err))
//			return
//		}
//		ref.MyWaitAdd("breakPoint")
//		breakPointObjectInit(m).NormalChanSendMsg("syMGetBreakPointTableList")
//		return
//	}
func missTableAction(k *SchedulePlan2) {
	event := "[miss]"
	log.MainLog().Info(fmt.Sprintf("(%v) %v Table %v.%v does not exist on the target side, and the table status is {missTable}", k.Object.ObjectLogSeq, event, k.Object.Schema, k.Object.Table))
	newTer.Write(k.TPod)
	k.XlsMiss.Data.Missing = GlobalPConfigs.dSns.SrcDBName
	newCell.Write(*k.XlsMiss)
	PlanContext.TaskBarSubsStatusSwap(k.subTaskInfo, "finish")
}
func sendStartMsgFinishFunc(_ string, _ any) (r any) {
	ref.MyWaitDel("schedulePlan")
	ref.MyWaitClose()
	PlanContext.objectEndMark(getCheckMod())
	return
}
func tableObjectErrorStatus(k *SchedulePlan2) bool {
	if atomic.LoadInt64(k.Status.ErrEnd) >= 0 {
		return false
	}
	newTer.Write(k.TPod)
	newCell.Write(k.XlsTerminal)
	newCell.Write(k.XlsResult)
	if GlobalPConfigs.result.Teletypewriter == "bar" {
		PlanContext.TaskBarSubsStatusSwap(k.subTaskInfo, "error")
	}
	return true
}
func sendStartMsgContinueFunc(_ string, f any) (r any) {
	_, chanParameter, _ := schemaMetaExecFuncParameterSplit(f.([]any))
	k := chanParameter.(*SchedulePlan2)
	if k.TPod.TableMiss {
		missTableAction(k)
		return true
	}
	return tableObjectErrorStatus(k)
}
func sendStartMsgExecFunc(_ string, f any) (r any) {
	_, chanParameter, _ := schemaMetaExecFuncParameterSplit(f.([]any))
	if funcName, ok := FuncRef[GlobalPConfigs.rules.CheckMode]; ok {
		if funcName.Kind() == reflect.Func {
			arguments := []reflect.Value{reflect.ValueOf(chanParameter)}
			funcName.Call(arguments)
		}
	} else {
		fmt.Println(fmt.Sprintf("-- [%v] %v report: checkObject parameter selection error, please refer to the log file for details, enable debug to get more information -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
		os.Exit(1)
	}
	return
}
func sendStartMsgInit() mq.MonitorMsgMq {
	var wg sync.WaitGroup
	ref.MyWaitAdd("schedulePlan")
	baseMsgSendBody = mq.NewMonitorMsgMq1().SetSendMsg(PlanContext.ChanSchedulePlanTable).SetCurryLimit(mq.SendCurryLimit{
		Wg:         &wg,
		Switch:     true,
		ManualDone: true,
		CurrySum:   make(chan struct{}, GlobalPConfigs.rules.Mtc),
	}).SetNormalQuitFunc(global.EFunc{
		FuncName: sendStartMsgFinishFunc}).SetContinueFunc(global.EFunc{FuncName: sendStartMsgContinueFunc}).SetExecFunc([]global.EFunc{
		{FuncName: sendStartMsgExecFunc}})
	return baseMsgSendBody
}

// sendStart
func schedulePlanStart1() {
	if err := PlanContext.Pool.Submit(func() {
		//发送消息
		fmt.Println(fmt.Sprintf("-- [%v] %v report: start schedule task %v -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName, GlobalPConfigs.rules.CheckMode))
		if sendStartMsgInit().NormalChanSendMsg("syMSchedulePlanStart1")!= nil {
			return
		}
		fmt.Println(fmt.Sprintf("-- [%v] %v report: schedule task meta finish -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
	}); err != nil {
		return
	}
}
func (context *SchedulePlanContext) monitor() {
	switch GlobalPConfigs.result.Teletypewriter {
	case "bar":
		context.TaskBarTable.BarMonitor()
	case "topic":
		monitorListener.TurnMonitor()
	default:
		context.TaskBarTable.BarMonitor()
	}
}
func (context *SchedulePlanContext) closeMonitor() {
	context.Close()
	//关闭消费者 阻塞
	consumer.Shutdown()
	//关闭生产者
	producer.Shutdown()
	newTer.Stop()
	newCell.Stop()
}
func (context *SchedulePlanContext) endClose() {
	if err := xlsFile.Close(); err != nil {
		return
	}
	closeObject.CloseAction()
	fmt.Println("")
	//输出结果信息
	fmt.Println("Check end")
	log.MainLog().Info(fmt.Sprintf("%v check object {%v} complete !!!", arg.ToolName, GlobalPConfigs.rules.CheckMode))
	fmt.Println(fmt.Sprintf("** %v Overview of results **", arg.ToolName))
	fmt.Println("Check time: ", fmt.Sprintf("%.2fs", time.Since(context.BeginTime).Seconds()), "(Seconds)")
	if err := newTer.Load("execTime", fmt.Sprintf("%v", time.Since(context.BeginTime).Seconds())); err != nil {
		log.ErrorLog().Error(fmt.Sprintf("%v", err))
	}
	newTer.Print()
}
func (context *SchedulePlanContext) Start(wg *sync.WaitGroup) {
	fmt.Println(fmt.Sprintf("-- [%v] %v report: start schedule task monitor -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
	terminal(wg)
	context.monitor()
	Cell(wg)
}

func (context *SchedulePlanContext) Check() {
	var wg sync.WaitGroup
	context.Start(&wg)
	closeObject.AbnormalExit()
	schedulePlanStart1()
	ref.MyWaitWait()
	context.closeMonitor()
	wg.Wait()
	context.endClose()
}

// Audit input
func Audit(m arg.ConfigParameter) {
	fmt.Println(fmt.Sprintf("-- [%v] %v report: begin input struct mode -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
	ref.MyWaitInit()
	schemaContext(&SchemaInitContextConfig{
		GtCheckConfig: m,
	}).Check()
	schedulePlanContext(&SchedulePlanContextConfig{
		GtCheckConfig: m,
		MQConfig:      cp.NewDefaultMq(),
		BeginTime:     m.BeginTime,
	}).Check()
}
